package com.hliushi.spark.sql

import java.lang

import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/18 19:27
 */
class TypeTransformation {

  // 1.创建SparkSession
  val spark: SparkSession = SparkSession.builder()
    .appName("type_transformation")
    .master("local[6]")
    .getOrCreate()

  import spark.implicits._

  @Test
  def trans(): Unit = {
    // 2.创建数据集
    val ds1 = Seq("hello spark", "hello hadoop").toDS()

    // 3.flatMap
    ds1.flatMap(x => x.split(" ")).show()

    // 4.Map
    val ds2 = Seq(Person("zhangsan", 15), Person("lisi", 20)).toDS()
    ds2.map((per: Person) => Person(per.name, per.age * 2)).show()

    // 5.mapPartitions
    /**
     * mapPartitions和map一样, 但是map的处理单位是每条数据, mapPartitions的处理单位是每个分区
     * def mapPartitions[U : Encoder](func: Iterator[T] => Iterator[U]): Dataset[U]
     */
    ds2.mapPartitions(
      // iter不能大到每个Executor的内存放不下, 不然就会OOM
      // 对每个元素进行转换, 后生成一个新的集合
      (iter: Iterator[Person]) => {
        // 这里iter.map() 方法是使用scala里面的方法, 不是RDD的map算子
        val result = iter.map((per: Person) => Person(per.name, per.age * 3))
        result
      }).show()
  }


  /**
   * map和mapPartitions是针对数据, 而transform是针对整个数据集
   * 这种方式最大的区别就是transform可以直接拿到Dataset进行操作
   * .
   * transform函数原型
   * def transform[U](t: Dataset[T] => Dataset[U]): Dataset[U]
   * #
   * range函数原型
   * def range(end: Long): Dataset[java.lang.Long] = range(0, end)
   */
  @Test
  def transform(): Unit = {
    val ds = spark.range(10)

    val ds2 = ds.transform((dataset: Dataset[lang.Long]) => dataset.withColumn("doubled", 'id * 2))

    ds2.show()
  }


  /**
   * as[Type]算子的主要作用是将弱类型的[DataFrame]转为强类型的Dataset
   * 它有很多适用场景, 最常见的还是在DataFrameReader框架读取数据的时候,
   * 因为DataFrameReader体系大部分情况下是将读出来的数据转换为DataFrame的形式
   * 如果后续需要使用Dataset的强类型的API, 需要使用as[Type]完成准换
   * as函数原型
   * def as[U : Encoder]: Dataset[U] = Dataset[U](sparkSession, logicalPlan)
   */
  @Test
  def as(): Unit = {
    // 1.定义schema
    val schema = StructType(
      List(
        StructField("name", StringType),
        StructField("age", IntegerType),
        StructField("gpa", FloatType)
      )
    )
    // 2.读取
    val df: DataFrame = spark.read
      .schema(schema)
      .option("delimiter", value = "\t")
      .csv("dataset/studenttab10k")

    val df02: Dataset[Row] = spark.read
      .schema(schema)
      .option("delimiter", value = "\t")
      .csv("dataset/studenttab10k")

    // 3.转换
    /**
     * 本质上: Dataset[Row].as[Student] => Dataset[Student]
     *
     * Dataset[(String, int, float)].as[Student] => Dataset[Student]
     */
    val ds = df.as[Student]

    // 4.输出
    ds.show()
  }


  /**
   * def filter(func: T => Boolean): Dataset[T]
   * def filter(condition: Column): Dataset[T]
   * def filter(conditionExpr: String)
   */
  @Test
  def filter(): Unit = {
    val ds = Seq(Person("zhangsan", 15), Person("lisi", 20)).toDS()
    val df = Seq(Person("zhangsan", 15), Person("lisi", 20)).toDF()

    /**
     * 一个是使用Dataset的filter, 另一个是使用DataFrame的filter
     */
    ds.filter((per: Person) => per.age > 15).show()
    df.filter('age > 15).show()
    df.filter("age > 15").show()
  }


  /**
   * groupByKey算子的返回结果是KeyValueGroupedDataset, 而不是一个Dataset
   * 所以必须要先经过KeyValueGroupedDataset中的方法进行聚合, 再转回Dataset
   * 才能使用Action得出结果,  --> 这也印证了分组后必须聚合的道理
   * #
   * groupByKey函数原型
   * def groupByKey[K: Encoder](func: T => K): KeyValueGroupedDataset[K, T]
   */
  @Test
  def groupByKey(): Unit = {
    val ds = Seq(Person("zhangsan", 15), Person("lisi", 20), Person("zhangsan", 21)).toDS()
    val df = Seq(Person("zhangsan", 15), Person("lisi", 20), Person("zhangsan", 21)).toDF()


    // select count(1) from person group by name
    val grouped = ds.groupByKey((per: Person) => per.name)

    val result = grouped.count()

    result.show()

    //  +--------+--------+
    //  |   value|count(1)|
    //  +--------+--------+
    //  |zhangsan|       2|
    //  |    lisi|       1|
    //  +--------+--------+
  }


  /**
   * randomSplit会按照传入的权重随机将一个Dataset分为多个Dataset
   * 传入的randomSplit的数组有多少个权重, 最终就会生成多少个Dataset, 这些权重的加倍和应为1, 否则将被标准化
   * randomSplit函数原型
   * def randomSplit(weights: Array[Double]): Array[Dataset[T]]
   * #
   * sample会随机在Dataset中抽样
   * sample函数原型
   * def sample(withReplacement: Boolean, fraction: Double): Dataset[T]
   */
  @Test
  def split(): Unit = {
    val ds = spark.range(15)
    // randomSplit, 切多少分, 权重多少
    val datasets = ds.randomSplit(Array(5, 2, 3))

    datasets.foreach(x => x.show())

    // sample
    ds.sample(withReplacement = false, 0.5).show()
  }


  /**
   * 其实orderBy 是 sort的别名, 所以它们所实现的功能是一样的
   * def orderBy(sortCol: String, sortCols: String*): Dataset[T] = sort(sortCol, sortCols : _*)
   * #
   * def orderBy(sortExprs: Column*): Dataset[T] = sort(sortExprs : _*)
   * #
   * def sort(sortCol: String, sortCols: String*): Dataset[T]
   * #
   * def sort(sortExprs: Column*): Dataset[T]
   */
  @Test
  def sort(): Unit = {
    val ds = Seq(Person("zhangsan", 15), Person("lisi", 20), Person("wangwu", 21)).toDS()

    ds.orderBy('name.desc).show()

    ds.sort($"age".asc).show()
  }


  /**
   * 去重
   * def distinct(): Dataset[T] = dropDuplicates()
   * #
   * def dropDuplicates(col1: String, cols: String*): Dataset[T]
   * #
   * def dropDuplicates(colNames: Array[String]): Dataset[T]
   * #
   * def dropDuplicates(colNames: Seq[String]): Dataset[T]
   * #
   * def dropDuplicates(): Dataset[T]
   */
  @Test
  def dropDuplicates(): Unit = {
    val ds = Seq(Person("zhangsan", 15), Person("lisi", 20), Person("zhangsan", 15)).toDS()

    // def distinct(): Dataset[T] = dropDuplicates()
    // 可以看出distinct就是 dropDuplicates方法的别名
    ds.distinct().show()


    /**
     * 当dropDuplicates可以去掉某一列中重复的行
     * 当dropDuplicates中没有传入列名的时候, 其含义是根据所有列去重,
     * dropDuplicates()方法还有一个别名, 叫做distinct
     */
    ds.dropDuplicates("age").show()

  }


  /**
   * 差集
   * def except(other: Dataset[T]): Dataset[T]
   * 交集
   * def intersect(other: Dataset[T]): Dataset[T]
   * 并集
   * def union(other: Dataset[T]): Dataset[T]
   * limit
   * def limit(n: Int): Dataset[T]
   */
  @Test
  def collection(): Unit = {
    val ds1 = spark.range(1, 10)
    val ds2 = spark.range(5, 15)

    // 差集     1, 2, 3, 4
    ds1.except(ds2).show()

    // 交集 [两个集合相交的地方]     5, 6, 7, 8, 9
    ds1.intersect(ds2).show()

    // 并集  1, 2, 3, 4, 5, ... 14 且去除了重复元素
    ds1.union(ds2).show()

    // limit
    ds1.limit(3).show()
  }

}

